package com.iakuil.app.common.redis;

import com.iakuil.app.common.core.util.ObjectUtils;
import com.iakuil.app.common.core.util.SpringUtils;
import com.iakuil.app.common.redis.base.RedisStreamListener;
import com.iakuil.app.common.redis.base.RedisTopicListener;
import com.iakuil.app.common.redis.utils.StreamUtils;
import com.iakuil.app.common.tp.ThreadPoolAutoConfiguration;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.Map;
import java.util.concurrent.Executor;

/**
 * Redis MessageListener配置
 *
 * @author Kai
 */
@Slf4j
@AutoConfiguration
@ConditionalOnClass(value = {RedisAutoConfiguration.class, ThreadPoolAutoConfiguration.class})
public class RedisListenerAutoConfiguration {

    @Resource(name = "redisListenerPollingExecutor")
    Executor threadPoolTaskExecutor;

    /**
     * 通过Redis Pub/Sub单播或广播
     *
     * <p>支持集群，非可靠投递，消息无序。
     */
    @Bean
    @ConditionalOnMissingBean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory factory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);

        // 自动扫描并注册MessageListener
        Map<String, RedisTopicListener> messageSubscribers = SpringUtils.getBeansOfType(RedisTopicListener.class);
        if (messageSubscribers != null && !messageSubscribers.isEmpty()) {
            for (Map.Entry<String, RedisTopicListener> entry : messageSubscribers.entrySet()) {
                RedisTopicListener messageSubscriber = entry.getValue();
                container.addMessageListener(new MessageListenerAdapter(messageSubscriber), messageSubscriber.getTopic());
            }
        }
        return container;
    }

    /**
     * 基于Redis Stream模式的ListenerContainer配置
     */
    @Bean
    @ConditionalOnMissingBean
    @SuppressWarnings({"rawtypes", "unchecked"})
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(RedisConnectionFactory factory) {
        // 创建Stream消息监听容器配置
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
            // 设置阻塞时间
            .pollTimeout(Duration.ofMillis(100))
            // 执行消息轮询的执行器
            .executor(threadPoolTaskExecutor)
            // 序列化器，统一使用String
            .serializer(new StringRedisSerializer())
            // 每批次记录数量
            .batchSize(1000)
            // 超时时间，设置为0，表示不超时（超时后会抛出异常）
//            .pollTimeout(Duration.ZERO)
            // 消息异常处理，仅打印
            .errorHandler(t -> log.error("[MQ exception] " + t.getMessage())).build();

        // 创建Stream消息监听容器
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = StreamMessageListenerContainer.create(factory, options);

        // 自动扫描并注册MessageListener
        Map<String, RedisStreamListener> streamListeners = SpringUtils.getBeansOfType(RedisStreamListener.class);
        if (ObjectUtils.isNotEmptyOrNull(streamListeners)) {
            for (Map.Entry<String, RedisStreamListener> entry : streamListeners.entrySet()) {
                listenerContainer.register(entry.getValue().getRequest(), entry.getValue());
                initGroup(entry.getValue().key(), entry.getValue().group());
            }
        }

        listenerContainer.start();
        return listenerContainer;
    }

    private void initGroup(String key, String group) {
        try {
            StreamUtils.createGroup(key, group);
            log.info("Redis Stream [{}]的消费组[{}]创建成功！", key, group);
            // 限制消息长度
            StreamUtils.trim(key, 10000);
        } catch (Exception e) {
            log.info("Redis Stream Group [{}]已存在，无需重复创建！", key);
        }
    }
}
